package com.junzhiit.nengji.cms.rabbitmq;

import java.util.List;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;

import com.junzhiit.nengji.cms.service.impl.BaseServiceImpl;
import com.junzhiit.nengji.dao.RabbitMQFailureDAO;
import com.junzhiit.nengji.model.RabbitMQFailure;

import cn.hutool.json.JSONUtil;

/**
 * @类简称 RabbitMQ框架 消息队列服务的一些公共代码实现
 * @功能点 1. 发送消息至队列服务器的接口
 * 2. 消息发送失败后将会把失败的消息新增指定的表内, 后续定时把表内消息记录再次发送（应用场景：消息队列服务器宕机）
 * 
 * @作者 Dave.lai
 */
public class BaseRabbitMQService extends BaseServiceImpl implements ConfirmCallback, ReturnCallback{
	
	private static final String DEMARCATION = ","; // 分割点
	
	@Autowired
	protected RabbitMQFailureDAO rabbitMQFailureDAO;
	@Autowired
	private RabbitTemplate rabbitTemplate;
	
	@Override
	public void init() {
		super.init();
		rabbitTemplate.setConfirmCallback(this);
		rabbitTemplate.setReturnCallback(this);
	}

	/**
	 * 	 将消息内容推至 RabbitMQ 服务器的指定队列中
	 * @param queueName 队列名称
	 * @param content 消息内容
	 */
	@Async
	public void toQueue(String queueName, String id, String content) {
		try {
			CorrelationData CorrelationData = new CorrelationData(id + DEMARCATION + queueName + DEMARCATION + content); // 把要发送的消息数据都合并上, 为后续出现发送失败的消息做重发操作
			MessagePostProcessor mpp = new MessagePostProcessor() {
				@Override
				public Message postProcessMessage(Message message) throws AmqpException {
					message.getMessageProperties().setCorrelationId(id.getBytes());
					return message;
				}
			};
			rabbitTemplate.convertAndSend(queueName, (Object) content, mpp, CorrelationData);
		} catch (Exception e) {
			logger.error("发送消息至'"+ queueName +"'队列异常！插入到消息队列失败表, 避免消息丢失");
			logger.error(JSONUtil.toJsonStr(e));
			e.printStackTrace();
			this.sendFailure(queueName, id, content);
		}
	}
	
	/**
	 * 	如果消息未能写入到队列, 则会触发该方法
	 * 	（1.发送消息的队列名跟配置里的队列名没对应上；2.消息队列服务器拒绝该消息插入到队列里）
	 */
	@Override
	public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
		logger.error("消息写入消息队列服务失败 ！插入到消息队列失败表, 避免消息丢失");
		String messageId = new String(message.getMessageProperties().getCorrelationId());
		String content = new String(message.getBody());
		this.sendFailure(routingKey, messageId, content);
	}

	/**
	 * 	只确认生产者消息发送至消费服务器成功; 消费者消费成功或已把消息写进队列等待消费这些都不保证
	 * 	（例如网络抖动、服务器队列拒绝接收啊等原因）
	 */
	@Override
	public void confirm(CorrelationData correlationData, boolean ack, String cause) {
		if (!ack) {
			logger.error("消息发送至 RabbitMQ 服务器失败！插入到消息队列失败表, 避免消息丢失");
			String[] correlationdata = correlationData.getId().split(DEMARCATION);
			String messageId = correlationdata[0];
			String queueName = correlationdata[1];
			String content = "";
			for (int i = 2; i < correlationdata.length; i++) {
				content += correlationdata[i];
				if(i < correlationdata.length - 1) {
					content += DEMARCATION;
				}
			}
			
			this.sendFailure(queueName, messageId, content);
        }
	}
	
	/**
	 * 	未能发送到消息队列服务器的消息将会插入失败表中，让定时器重新发送
	 * @param queueName 队列名称
	 * @param messageId 消息对应表的id
	 * @param content 消息内容
	 */
	public void sendFailure(String queueName, String messageId, String content) {
		try {
			RabbitMQFailure msg = new RabbitMQFailure();
			msg.setQueueName(queueName);
			msg.setMessageId(messageId);
			msg.setContent(content);
			rabbitMQFailureDAO.insert(msg);
		} catch (Exception e) {
			logger.error("消息队列服务：录入失败消息记录至数据库异常, 可能是重复数据触发唯一约束, 如果是, 则捕获到后不用处理");
			logger.error(JSONUtil.toJsonStr(e));
			e.printStackTrace();
		}
	}
	
	/**
	 * 	定时任务执行该方法，对失败的消息进行重发
	 */
	public void retrySendFailureMessage() {
		List<RabbitMQFailure> failures = rabbitMQFailureDAO.top200();
		for (RabbitMQFailure msg : failures) {
			rabbitMQFailureDAO.delete(msg.getId());
			toQueue(msg.getQueueName(), msg.getMessageId(), msg.getContent());
		}
	}
}
